package com.navi.service.impl;

import com.navi.GetIP;
import com.navi.entity.TableEntity;
import com.navi.util.LogUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.sql.SQLException;
import java.sql.Statement;
import java.util.List;
import java.util.stream.Collectors;

/**
 * Created by ghost on 2018/11/19.
 */
@Service
public class ExtTableOperationService {

    @Value("${stagetable.schema}")
    private String stageTableSchema;
    @Value("${external.table.schema}")
    private String extTableSchema;
    @Value("${gpfdist.host}")
    private String gpfdistHost;
    @Value("${gpfdist.port}")
    private String gpfdistPort;


    @Autowired
    private TableMetaService tableMetaService;

    private Logger logger = LoggerFactory.getLogger(this.getClass());





    /**
     * 由于stage table的batch update速度太慢，需要使用gpfdist从外部表一次写入
     * <p>
     * 注:创建外部表时应该 指定 null 类型关键字，以免带来不必要的麻烦，比如数字型和时间类型 需要转换的问题
     *
     * @param stmt
     * @param extTableName
     * @param tableEntity
     * @return
     */
    public void createExtTable(Statement stmt, String extTableName, TableEntity tableEntity) throws SQLException {
        logger.info("get stageTableSchema [{}] info ... ", extTableName);
        final TableEntity extTableEntity = tableMetaService.getTableMetaInfo(extTableSchema, extTableName, stmt);
        //已存在的 ext table和即将要创建的 ext table一样时，直接使用，不再重新创建
//        boolean isSame = compare(extTableEntity, tableEntity);
        boolean isSame = extTableEntity != null && compare(extTableEntity, tableEntity);


        if (isSame) {
            logger.info("ext table {}.{} already exist. ", extTableSchema, extTableName);
        } else {
            String dropSql = "drop external table if exists " + extTableSchema + "." + extTableName;
            final String fileName = String.format("%s.txt", tableEntity.getTableName());

            //获取本机IP
            GetIP local = new GetIP();
            String localIP = local.getCurrentIp();
//            String location = String.format("gpfdist://%s:%s/%s", gpfdistHost, gpfdistPort, fileName);
            String location = String.format("gpfdist://%s:%s/%s", localIP, gpfdistPort, fileName);

            String columnString = tableEntity.getColumnEntityList().stream().map(cl ->
                    cl.getColumnName() + " " + cl.getDataType()
            ).collect(Collectors.joining(","));

            StringBuffer createTableBuffer = new StringBuffer("create external table ");
            createTableBuffer.append(extTableSchema).append(".").append(extTableName);
            createTableBuffer.append("(").append(columnString).append(")");
            createTableBuffer.append(" location ('").append(location).append("").append("')");
            //注意此处的null ''
            createTableBuffer.append(" FORMAT 'text' (delimiter '^' null '' escape '\\\\')");
            createTableBuffer.append(" ENCODING 'UTF8';");

            try {
                logger.info("drop external table sql : {}", dropSql);
                stmt.execute(dropSql);
                logger.info("create external table sql : {}", createTableBuffer.toString());
                stmt.executeUpdate(createTableBuffer.toString());
                logger.info("created external table : {}.{}", extTableSchema, extTableName);
                tableMetaService.saveTableMetaInfo2Map(extTableSchema, extTableName, stmt);
            } catch (SQLException e) {
                logger.error("created external table : {}.{}, failed: {}", extTableSchema, extTableName, e.toString());
                LogUtil.pringStackTrace(e);
            }
        }
    }


    /*
    比较 EXT 表
    true 如果存在相应（相同）的表结构缓存
     */
    public boolean compare(TableEntity existStageTableDdl, TableEntity tableEntity) {
        logger.info("comparing table meta [{}] ... ", existStageTableDdl);
        final List<Boolean> resultList = tableEntity.getColumnEntityList().stream().map(
                c1 -> existStageTableDdl.getColumnEntityList().parallelStream().anyMatch(
                        c2 -> c1.getColumnName().equals(c2.getColumnName()) && c1.getDataType().equals(c2.getDataType())
                )
        ).collect(Collectors.toList());
        boolean res=resultList.stream().anyMatch(r -> r == false) ? false : true;
        return res;
    }
}
